package io.kiki.sba.registry.server.shared.remoting;

import io.kiki.sba.registry.api.Callback;
import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.api.ChannelHandler;
import io.kiki.sba.registry.api.Client;
import io.kiki.sba.registry.api.exchange.NodeExchanger;
import io.kiki.sba.registry.api.exchange.RequestException;
import io.kiki.sba.registry.api.exchange.message.Request;
import io.kiki.sba.registry.api.exchange.message.Response;
import io.kiki.sba.registry.api.exchange.message.SimpleRequest;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.remoting.netty.NettyClient;
import io.kiki.sba.registry.remoting.netty.NettyDataExchanger;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;

import java.util.*;


public abstract class ClientSideExchanger implements NodeExchanger, ApplicationListener<ContextRefreshedEvent> {
    private static final Logger logger = LoggerFactory.getLogger(ClientSideExchanger.class);
    protected final String serverType;
    private final Connector connector;
    @Autowired
    protected NettyDataExchanger nettyDataExchanger;
    protected volatile Set<String> serverIpSet = new HashSet<>();

    protected ClientSideExchanger(String serverType) {
        this.serverType = serverType;
        this.connector = new Connector();
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        init();
    }

    public void init() {
        ConcurrentUtils.createDaemonThread(serverType + "-async-connector", connector).start();
        logger.info("{} init connector", serverType);
    }

    @Override
    public Response request(Request request) throws RequestException {
        final Url url = request.getRequestUrl();
        if (url == null) {
            throw new RequestException("null url", request);
        }
        Client client = nettyDataExchanger.getClient(serverType);
        final int timeout = request.getTimeout() != null ? request.getTimeout() : getRpcTimeoutMillis();
        try {
            if (client == null) {
                connect(url);
                client = nettyDataExchanger.getClient(serverType);
            }
            Callback callback = request.getCallBack();
            if (callback == null) {
                final Object result = client.sendSync(url, request.getRequestBody(), timeout);
                return () -> result;
            } else {
                client.sendWithCallback(url, request.getRequestBody(), callback, timeout);
                return () -> Response.ResultStatus.SUCCESSFUL;
            }
        } catch (Throwable e) {
            throw new RequestException(serverType + "Exchanger request error! Request url:" + url, request, e);
        }
    }

    public Response requestRaw(String ip, Object raw) throws RequestException {
        Request req = new SimpleRequest(raw, new Url(ip, getServerPort()));
        return request(req);
    }

    @Override
    public Client connectServer() {
        Set<String> serverIpSet = this.serverIpSet;
        if (!serverIpSet.isEmpty()) {
            int count = tryConnectAllServer(serverIpSet);
            if (count == 0) {
                throw new RuntimeException("failed to connect any servers, " + serverIpSet);
            }
        }
        return getClient();
    }

    public Client getClient() {
        return nettyDataExchanger.getClient(serverType);
    }

    protected int tryConnectAllServer(Set<String> ips) {
        int connectCount = 0;
        for (String ip : ips) {
            Url url = new Url(ip, getServerPort());
            try {
                connect(url);
                connectCount++;
            } catch (Throwable e) {
                logger.error("Exchanger connect server error!url:" + url, e);
            }
        }
        return connectCount;
    }

    public Channel connect(Url url) {
        Client client = getClient();
        if (client == null) {
            synchronized (this) {
                client = getClient();
                if (client == null) {
                    client = nettyDataExchanger.connect(serverType, url, getConnNum(), getClientHandlers().toArray(new ChannelHandler[0]));
                }
            }
        }
        Channel channel = client.getChannel(url);
        if (channel == null) {
            synchronized (this) {
                channel = client.getChannel(url);
                if (channel == null) {
                    channel = client.connect(url);
                }
            }
        }
        return channel;
    }

    public Map<String, List<io.kiki.stack.netty.channelManager.Channel>> getConnections() {
        Client client = nettyDataExchanger.getClient(serverType);
        if (client == null) {
            return Collections.emptyMap();
        }
        return ((NettyClient) client).getConnections();
    }

    public void notifyConnectServerAsync() {
        connector.wakeup();
    }

    public abstract int getRpcTimeoutMillis();

    public abstract int getServerPort();

    public int getConnNum() {
        return 1;
    }

    protected abstract Collection<ChannelHandler> getClientHandlers();

    public synchronized void setServerIpSet(Collection<String> serverIpSet) {
        this.serverIpSet = Collections.unmodifiableSet(new HashSet<>(serverIpSet));
    }


    private final class Connector extends WakeUpLoopExecuteTask {

        @Override
        public void _execute_() {
            Set<String> ips = serverIpSet;
            try {
                tryConnectAllServer(ips);
            } catch (Throwable e) {
                logger.error("failded to connect {}", ips, e);
            }
        }

        @Override
        public int getWaitingMillis() {
            return 3000;
        }
    }
}
